# -*- coding: utf-8 -*-
"""
项目名称：NoSql_kafka_project
文件名称：main.py
创建时间：2025-06-05 16:21:50

系统用户：Administrator
作　　者：無以菱
联系邮箱：huangjing2001.guet@qq.com
功能描述：
- 功能：协调各组件运行
- 线程1：Kafka生产者（后台）
- 线程2：Kafka消费者（后台）
- 线程3：Spark分析任务（后台，定时执行）
- 主线程：Flask Web应用
"""
import threading  
import time  
import sys 
import logging  
from kafka_producer import run_kafka_producer  # 导入Kafka生产者函数
from kafka_consumer import run_kafka_consumer  # 导入Kafka消费者函数
from spark_hbase_redis import run_spark_analysis  # 导入Spark分析函数
from flask_app import run_flask_app  # 导入Flask应用函数

# 配置日志
# 设置日志级别为INFO，格式包含时间、名称、级别和消息
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 获取当前模块的logger
logger = logging.getLogger(__name__)


def start_kafka_producer():
    """
    启动Kafka生产者线程
    
    该函数作为线程目标，启动Kafka生产者，生成模拟订单数据
    """
    try:
        logger.info("正在启动Kafka生产者线程...")
        # 运行Kafka生产者，该函数包含无限循环，会持续生成数据
        run_kafka_producer()
    except Exception as e:
        # 捕获并记录生产者线程中的异常
        logger.error(f"Kafka生产者线程出错: {e}")


def start_kafka_consumer():
    """
    启动Kafka消费者线程
    
    该函数作为线程目标，启动Kafka消费者，处理订单数据并写入HBase
    """
    try:
        logger.info("正在启动Kafka消费者线程...")
        # 运行Kafka消费者，该函数包含无限循环，会持续消费数据
        run_kafka_consumer()
    except Exception as e:
        # 捕获并记录消费者线程中的异常
        logger.error(f"Kafka消费者线程出错: {e}")


def spark_scheduler():
    """
    定期执行Spark分析任务
    
    该函数作为线程目标，每5秒执行一次Spark分析任务，分析HBase中的数据并将结果存入Redis
    """
    logger.info("Spark分析调度器启动")
    while True:  # 无限循环，持续执行分析任务
        try:
            # 执行Spark分析任务
            run_spark_analysis()
        except Exception as e:
            # 捕获并记录分析任务中的异常
            logger.error(f"Spark分析任务执行错误: {e}")

        # 等待5秒后再次执行
        logger.debug("等待5秒后执行下一次分析...")
        time.sleep(5)  # 暂停5秒


def main():
    """
    主函数，启动所有组件
    
    协调各组件的启动和运行，包括：
    1. 启动Kafka生产者线程
    2. 启动Kafka消费者线程
    3. 启动Spark分析线程
    4. 在主线程中运行Flask Web应用
    """
    try:
        # 启动Kafka生产者（后台线程）
        # daemon=True表示这是一个守护线程，当主线程结束时，该线程会自动终止
        kafka_producer_thread = threading.Thread(target=start_kafka_producer, daemon=True)
        kafka_producer_thread.start()  # 启动线程
        logger.info("Kafka生产者线程已启动")

        # 启动Kafka消费者（后台线程）
        kafka_consumer_thread = threading.Thread(target=start_kafka_consumer, daemon=True)
        kafka_consumer_thread.start()  # 启动线程
        logger.info("Kafka消费者线程已启动")

        # 启动Spark统计任务（后台线程）
        spark_thread = threading.Thread(target=spark_scheduler, daemon=True)
        spark_thread.start()  # 启动线程
        logger.info("Spark分析线程已启动")

        # 等待线程初始化
        # 给线程一些时间来初始化，确保它们正常运行
        time.sleep(2)

        # 检查线程状态
        # 确认所有线程都在正常运行
        threads = [
            ("Kafka生产者", kafka_producer_thread),
            ("Kafka消费者", kafka_consumer_thread),
            ("Spark分析", spark_thread)
        ]

        for name, thread in threads:
            if thread.is_alive():
                # 线程正在运行
                logger.info(f"{name}线程运行正常")
            else:
                # 线程未启动或已终止
                logger.error(f"{name}线程未正常运行")

        # 启动Flask应用（主线程）
        # Flask应用在主线程中运行，会阻塞直到应用停止
        logger.info("启动Flask Web应用...")
        run_flask_app()

    except KeyboardInterrupt:
        # 处理键盘中断（Ctrl+C）
        logger.info("程序被用户中断")
    except Exception as e:
        # 处理其他未预期的异常
        logger.error(f"程序执行过程中出错: {e}")
        sys.exit(1)  # 非正常退出，返回错误码1


if __name__ == "__main__":
    # 单独运行此文件时执行
    logger.info("正在启动实时支付数据分析系统...")
    main()  # 执行主函数